#!/usr/bin/perl
# 2012/11/07 Sorting map file

use strict;
use warnings;
use IO::File;
use Getopt::Std;
use File::Basename;
use Carp qw(croak);
$| = 0;

if(@ARGV != 4 or $ARGV[0] eq '--help'){
	die "Usage : $0 -i [map file] -o [sorted outfile]\n";
}

my %num = (
	'I'	=> 100000000,
	'II'	=> 200000000,
	'III'	=> 300000000,
	'MT'	=> 400000000,
	'MTR'	=> 500000000,
	'AB325691'	=> 600000000,
	'NA'	=> 700000000
);

my $LINE_PER_FILES = 500000;

use threads;
use Thread::Queue;
use Thread::Semaphore;


#---------------------------------------
# map format
#---------------------------------------
# id chr1 loc1 direction1 quality1 hind1 hind_loc1 chr2 loc2 direction2 quality2 hind2 hind_loc2 type(unique or repeat) number(alignable locations) locations

my %opt;
getopts("i:o:", \%opt);
my $FILE_map = $opt{i};
my $FILE_sorted = $opt{o};
my ($name, $dir, $ext) = &fileparse($FILE_map, '\..*');

my $THREAD_NUMBER = 20;
my $semaphore = Thread::Semaphore->new($THREAD_NUMBER);
my $FILEQues = Thread::Queue->new();


my $FILE_NUMBER = 0;
my %data;
my $count = 0;

#---------------------------------------
# collect data
#---------------------------------------
my $fh_map = IO::File->new($FILE_map) or die "cannot open $FILE_map: $!";
while($_ = $fh_map->getline()){
	s/\r?\n//;
	my ($id, $chr1, $loc1, $direction1, $mapQ1, $hinID1, $hinLoc1, $chr2, $loc2, $direction2, $mapQ2, $hinID2, $hinLoc2, $type, $num, $locations) = split /\t/;

	if($loc1 eq 'NA'){
		next;
	}
	if($loc2 eq 'NA'){
		next;
	}
	my $value1 = $num{$chr1} + $loc1;
	my $value2 = $num{$chr2} + $loc2;
	my $value = $value1 * 1000000000 + $value2;


	my $key = "$value\t$_";
	$data{$key} = $value;
	$count++;

	if($count > $LINE_PER_FILES){
		$FILE_NUMBER++;
		my $tmpFile = $dir . 'tmp_' . $name . '_' . $FILE_NUMBER . '.txt';
		$semaphore->down();
		$FILEQues->enqueue($tmpFile);
		&status("output $tmpFile");
		my $thread = new threads(\&output_sort_file, $tmpFile, \%data);
		%data = ();
		$count = 0;
	}
}
$fh_map->close();

### 残りの行を処理
if($count > 0){
	$semaphore->down();
	$FILE_NUMBER++;
	my $tmpFile = $dir . 'tmp_' . $name . '_' . $FILE_NUMBER . '.txt';
	$FILEQues->enqueue($tmpFile);
	&status("output final $tmpFile");
	my $thread = new threads(\&output_sort_file, $tmpFile, \%data);
	%data = ();
}

sub output_sort_file{
	my ($file, $ref) = @_;
	my $fh_tmp = IO::File->new($file, 'w') or die "cannot write $file: $!";
	foreach my $k(sort {$ref->{$a} <=> $ref->{$b} } keys %$ref){
		$fh_tmp->print("$k\n");
	}
	$fh_tmp->close();
	threads->yield();
	$semaphore->up();
}

### merge to $THREAD_NUMBER files
&status("merge start");
my @threads;
#---------------------------------------
# sorting
#---------------------------------------
for(1..$THREAD_NUMBER){
    $semaphore->down();
    my $FILE_result = $dir . 'merged_' . $name . '_' . $_ . '.txt';
    my $FILE_tmp = $dir . 'merged_' . $name . '_' . $_ . '.tmp';

    my $FILE_first = $FILEQues->dequeue_nb();
    &status("read $FILE_first");

    # if no files in ques, stop
    unless($FILE_first){
    	last;
    }

    my $thread = new threads(\&merge_threads, $FILE_result, $FILE_tmp, $FILE_first);
    push @threads, $thread;
}

### wait until all thread will finish
foreach (@threads){
	$_->join();
}

sub merge_threads{
	my ($FILE_result, $FILE_tmp, $FILE_first) = @_;
	rename $FILE_first, $FILE_result or die "cannot rename $FILE_first to $FILE_result: $!";

	while(my $tmpFile = $FILEQues->dequeue_nb()){
		&status("read $tmpFile");
		my $fh_result = IO::File->new($FILE_result) or die "cannot open $FILE_result: $!";
		my $fh_tmp = IO::File->new($tmpFile) or die "cannot open $tmpFile: $!";
		my $fh_out = IO::File->new($FILE_tmp, 'w') or die "cannot write $FILE_tmp: $!";

		my $s1 = $fh_result->getline();
		my $s2 = $fh_tmp->getline();

		while($s1 && $s2){
			my ($n1, @other1) = split /\t/, $s1;
			my ($n2, @other2) = split /\t/, $s2;
			if($n1 < $n2){
				$fh_out->print("$s1");
				$s1 = $fh_result->getline();
			}elsif($n1 > $n2){
				$fh_out->print("$s2");
				$s2 = $fh_tmp->getline();
			}else{
				$fh_out->print("$s1");
				$fh_out->print("$s2");
				$s1 = $fh_result->getline();
				$s2 = $fh_tmp->getline();
			}
		}
		while($s1){
			$fh_out->print("$s1");
			$s1 = $fh_result->getline();
		}
		while($s2){
			$fh_out->print("$s2");
			$s2 = $fh_tmp->getline();
		}
		$fh_out->close();
		$fh_result->close();
		$fh_tmp->close();

		rename $FILE_tmp, $FILE_result or die "cannot rename file: $!";
		unlink $tmpFile;
	}
	$FILEQues->enqueue($FILE_result);
	&status("output process file $FILE_result");
	$semaphore->up();
}



### merge remaining files
{
	my $FILE_TMP_FINAL = $dir . 'final_' . $name . '.tmp';
	my $FILE_first = $FILEQues->dequeue();
	&status("read $FILE_first");
	rename $FILE_first, $FILE_sorted or die "cannot rename $FILE_first to $FILE_sorted: $!";
	while(my $tmpFile = $FILEQues->dequeue_nb()){
		&status("read $tmpFile");
		my $fh_result = IO::File->new($FILE_sorted) or die "cannot open $FILE_sorted: $!";
		my $fh_tmp = IO::File->new($tmpFile) or die "cannot open $tmpFile: $!";
		my $fh_out = IO::File->new($FILE_TMP_FINAL, 'w') or die "cannot write $FILE_TMP_FINAL: $!";

		my $s1 = $fh_result->getline();
		my $s2 = $fh_tmp->getline();

		while($s1 && $s2){
			my ($n1, @other1) = split /\t/, $s1;
			my ($n2, @other2) = split /\t/, $s2;
			if($n1 < $n2){
				$fh_out->print("$s1");
				$s1 = $fh_result->getline();
			}elsif($n1 > $n2){
				$fh_out->print("$s2");
				$s2 = $fh_tmp->getline();
			}else{
				$fh_out->print("$s1");
				$fh_out->print("$s2");
				$s1 = $fh_result->getline();
				$s2 = $fh_tmp->getline();
			}
		}
		while($s1){
			$fh_out->print("$s1");
			$s1 = $fh_result->getline();
		}
		while($s2){
			$fh_out->print("$s2");
			$s2 = $fh_tmp->getline();
		}
		$fh_out->close();
		$fh_result->close();
		$fh_tmp->close();

		rename $FILE_TMP_FINAL, $FILE_sorted or die "cannot rename file: $!";
		unlink $tmpFile;
	}
}


#---------------------------------------
# remove the number
#---------------------------------------
{
	my $FILE_tmp = $dir . 'removeNumber_' . $name . '.tmp';
	my $fh_out = IO::File->new($FILE_tmp, 'w') or die "cannot write tmp: $!";
	my $fh_in = IO::File->new($FILE_sorted) or die "cannot open $FILE_sorted: $!";

	while($_ = $fh_in->getline()){
		my ($n, $id, $chr1, $loc1, $direction1, $mapQ1, $hinID1, $hinLoc1, $chr2, $loc2, $direction2, $mapQ2, $hinID2, $hinLoc2, $type, $num, $locations) = split /\t/;
		my $key = join("\t", $id, $chr1, $loc1, $direction1, $mapQ1, $hinID1, $hinLoc1, $chr2, $loc2, $direction2, $mapQ2, $hinID2, $hinLoc2, $type, $num, $locations);
		$fh_out->print("$key");
	}
	$fh_in->close();
	$fh_out->close();
	rename $FILE_tmp, $FILE_sorted or die "cannot rename file: $!";
}

sub status{
	my ($message) = @_;
	my $time = &time();
	print "[$time] $message ... \n";
}


sub time{
	my ($sec, $min, $hour, $mday, $mon, $year) = localtime(time);
	my $time = sprintf ("%d/%0d %02d:%02d",$mon+1, $mday, $hour, $min);
	return $time;
}
